package Bestv.OTT_B2B_Replay;

import java.util.Map;
import org.apache.log4j.Logger;
import org.apache.storm.task.OutputCollector;
import org.apache.storm.task.TopologyContext;
import org.apache.storm.topology.OutputFieldsDeclarer;
import org.apache.storm.topology.base.BaseRichBolt;
import org.apache.storm.tuple.Fields;
import org.apache.storm.tuple.Tuple;
import org.apache.storm.tuple.Values;
import org.json.simple.JSONObject;

import com.esotericsoftware.minlog.Log;

public class NginxResolveBolt extends BaseRichBolt{
	
	private static final long serialVersionUID= -917956850826439496L;
	private OutputCollector collector;
	Logger logger;
	Nginx_Access_Resolve nginx_log;
	
	@Override
	public void execute(Tuple input) {
		// TODO Auto-generated method stub
		String KafkaInMessage = (String) input.getValue(0);
		//logger.info("原始日志是："+KafkaInMessage);
		nginx_log=new Nginx_Access_Resolve(KafkaInMessage);
		JSONObject log_json=nginx_log.nginx_resolve();
		//logger.info("处理后的日志json是： "+log_json.toString());
		//collector.emit(input,new Values(log_json.toString()));
		if ( log_json.size() > 0) {
			collector.emit(input,new Values(log_json)); 
	        collector.ack(input);
		}
		
	}

	@Override
	public void prepare(Map stormConf, TopologyContext context, OutputCollector collector) {
		// TODO Auto-generated method stub
		this.collector=collector;
		this.logger=Logger.getLogger(Nginx_Access_Resolve.class);
		
	}

	@Override
	public void declareOutputFields(OutputFieldsDeclarer declarer) {
		// TODO Auto-generated method stub
		declarer.declare(new Fields("log_json"));
		
	}

}
